JAVA 实现 Redis 发布订阅 |
您所在的位置：网站首页 › redis实现订阅发布 java › JAVA 实现 Redis 发布订阅 |
JAVA 实现 Redis 发布订阅
|
案例仓库地址
gitee
# Redis 发布订阅
发布订阅：消息发布者发布消息 和 消息订阅者接收消息，两者之间通过某种媒介联系起来 例如订杂志，当自己订阅了爱格杂志，每个月会发刊一本。到发布的时候派送员将杂志送到自己手上就能看到杂志内容。只有我们订阅了该杂志才会派送给我们 Redis 发布订阅（pub/sub）是一种 消息通信模式 ：发送者（pub）发送消息，订阅者（sub）接收消息。 Redis 客户端可以订阅任意数量的频道。 订阅 / 发布消息图：
图中可以看出，所需： 消息发送者 、 2. 频道 、 3. 消息订阅者 发布订阅机制 当一个客户端通过 PUBLISH 命令向订阅者发布消息的时候，称这个客户端为发布者publisher当一个客户端通过subscribe 或者 PSUBSCRIBE 接收消息时，称这个客户端为 订阅者 subscriber为了解耦发布者和订阅者之间的关系，Redis 使用了频道channel(频道)作为两者之间的中介，发布者直接把消息发送给 channel，而 channel 负责把消息发送给订阅者，发布者和订阅者之间没有直接的联系，都不知道对方的存在订阅者 1，2，3 订阅了频道 channel，当有消息发布给频道时，这个消息就会被发送到三个订阅者客户端
学习参考链接 引入依赖 org.springframework.boot spring-boot-starter-data-redis添加配置文件 spring: redis: host: 127.0.0.1 database: 5 password: port: 6379 Listener模式创建一个监听容器 @Configuration public class CatListenerConfig extends CachingConfigurerSupport { /** * 消息监听容器 * * @param factory * @return */ @Bean RedisMessageListenerContainer container(RedisConnectionFactory factory){ RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(factory); //订阅一个通道 该处的通道名是发布消息时的名称 container.setConnectionFactory(connectionFactory); //订阅了一个叫cat 的通道 container.addMessageListener(catAdapter, new PatternTopic("cat")); container.addMessageListener(fishAdapter, new PatternTopic("fish")); return container; } } @Component public class CatListener implements MessageListener { @Override public void onMessage(Message message, byte[] bytes) { System.out.println("我是监听者,我监听到的消息是 " + message.toString()); } } package com.maoxs.listener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.core.RedisTemplate; /** * 监听发送的消息 */ public class FishListener implements MessageListener { @Autowired RedisTemplate redisTemplate; @Override public void onMessage(Message message, byte[] bytes) { System.out.println("我是Fish监听" + message.toString()); } }建测试类，测试发布监听 @RestController public class TestController { @Resource StringRedisTemplate stringRedisTemplate; @PostMapping("/cat") public void test2(){ stringRedisTemplate.convertAndSend("cat","测试：消息发布者发布消息"); } @PostMapping("/fish") public void fish(){ stringRedisTemplate.convertAndSend("fish","测试：消息发布者发布消息"); } }测试结果
2.1 配置 /** * MessageListenerAdapter 模式 * 该处topic的 key为bean的name * @param connectionFactory * @param adapterMap * @return */ @Bean public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,Map adapterMap) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); adapterMap.keySet().forEach(topic-> container.addMessageListener(adapterMap.get(topic),new PatternTopic(topic))); return container; } 监听1 package com.sst.loan.risk.listener; import com.sst.loan.risk.manager.RuleLoadManager; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.stereotype.Component; import javax.annotation.Resource; /** * 监听适配器 * * @author 蔡定努 * @date 2023/06/13 10:26 */ @Slf4j @Component("ruleRefreshAdapter") public class RuleRefreshAdapter extends MessageListenerAdapter { @Override public void onMessage(Message message, byte[] bytes) { log.info(">>>>>>> 消息适配器收到刷新规则的请求 |
【本文地址】
今日新闻 |
推荐新闻 |